package com.sheep.api.hbase.dao.impl;

import com.sheep.api.bean.ADSBBean;
import com.sheep.api.hbase.dao.HbaseCommonDao;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Repository;

import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * User: Administrator
 * Date: 16-4-25
 * Time: 下午3:13
 * To change this template use File | Settings | File Templates.
 */
@Repository
public class HbaseCommonDaoImpl implements HbaseCommonDao {
    @Autowired
    private HbaseTemplate hbaseTemplate;

    private String tableName = "ADSB";

    public static byte[] ADSB_INFO = Bytes.toBytes("adsbInfo");

    private byte[] c_an = Bytes.toBytes("an");
    private byte[] c_fi = Bytes.toBytes("fi");
    private byte[] c_time = Bytes.toBytes("time");
    private byte[] c_lat = Bytes.toBytes("lat");
    private byte[] c_lon = Bytes.toBytes("lon");
    private byte[] c_vec = Bytes.toBytes("vec");
    private byte[] c_cas = Bytes.toBytes("cas");

    @Override
    public List<ADSBBean> findAll() {
        return hbaseTemplate.find(tableName, "cfInfo", new RowMapper<ADSBBean>() {
            @Override
            public ADSBBean mapRow(Result result, int rowNum) throws Exception {
                return new ADSBBean(Bytes.toLong(result.getValue(ADSB_INFO, c_time)),
                        Bytes.toString(result.getValue(ADSB_INFO, c_an)),
                        Bytes.toString(result.getValue(ADSB_INFO, c_fi)),
                        Bytes.toLong(result.getValue(ADSB_INFO, c_lat)),
                        Bytes.toLong(result.getValue(ADSB_INFO, c_lon)),
                        Bytes.toLong(result.getValue(ADSB_INFO, c_vec)),
                        Bytes.toLong(result.getValue(ADSB_INFO, c_cas)));
            }
        });

    }

    @Override
    public ADSBBean save(final long time, final String an, final String fi, final double lat, final double lon, final double vec, final double cas) {
        return hbaseTemplate.execute(tableName, new TableCallback<ADSBBean>() {
            public ADSBBean doInTable(HTableInterface table) throws Throwable {
                ADSBBean adsb = new ADSBBean(time, an, fi, lat, lon, vec, cas);
                Put p = new Put(Bytes.toBytes(adsb.getAn() + adsb.getFi()));
                p.add(ADSB_INFO, c_an, Bytes.toBytes(adsb.getAn()));
                p.add(ADSB_INFO, c_fi, Bytes.toBytes(adsb.getFi()));
                p.add(ADSB_INFO, c_time, Bytes.toBytes(adsb.getTime()));
                p.add(ADSB_INFO, c_lat, Bytes.toBytes(adsb.getLat()));
                p.add(ADSB_INFO, c_lon, Bytes.toBytes(adsb.getLon()));
                p.add(ADSB_INFO, c_cas, Bytes.toBytes(adsb.getCas()));
                p.add(ADSB_INFO, c_vec, Bytes.toBytes(adsb.getVec()));
                table.put(p);
                return adsb;
            }
        });
    }
}
